package com.mqstudy.kafka.producer;

import com.mqstudy.common.model.Message;
import com.mqstudy.kafka.config.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.CompletableFuture;

/**
 * Kafka生产者
 */
@Slf4j
@Component
public class KafkaProducer {
    
    @Autowired
    private KafkaTemplate<String, Message> kafkaTemplate;
    
    /**
     * 发送消息到测试主题
     */
    public void sendTestMessage(Message message) {
        log.info("发送测试消息: {}", message);
        sendMessage(KafkaConfig.TEST_TOPIC, message.getMessageId(), message);
    }
    
    /**
     * 发送订单消息
     */
    public void sendOrderMessage(Message message) {
        log.info("发送订单消息: {}", message);
        sendMessage(KafkaConfig.ORDER_TOPIC, message.getKey(), message);
    }
    
    /**
     * 发送用户消息
     */
    public void sendUserMessage(Message message) {
        log.info("发送用户消息: {}", message);
        sendMessage(KafkaConfig.USER_TOPIC, message.getKey(), message);
    }
    
    /**
     * 发送日志消息
     */
    public void sendLogMessage(Message message) {
        log.info("发送日志消息: {}", message);
        sendMessage(KafkaConfig.LOG_TOPIC, message.getMessageId(), message);
    }
    
    /**
     * 发送流式消息
     */
    public void sendStreamMessage(Message message) {
        log.info("发送流式消息: {}", message);
        sendMessage(KafkaConfig.STREAM_TOPIC, message.getMessageId(), message);
    }
    
    /**
     * 发送消息的通用方法
     */
    private void sendMessage(String topic, String key, Message message) {
        ListenableFuture<SendResult<String, Message>> future = kafkaTemplate.send(topic, key, message);
        
        future.addCallback(new ListenableFutureCallback<SendResult<String, Message>>() {
            @Override
            public void onSuccess(SendResult<String, Message> result) {
                log.info("消息发送成功 - Topic: {}, Partition: {}, Offset: {}, Key: {}", 
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset(),
                        key);
            }
            
            @Override
            public void onFailure(Throwable ex) {
                log.error("消息发送失败 - Topic: {}, Key: {}, Error: {}", topic, key, ex.getMessage(), ex);
            }
        });
    }
    
    /**
     * 异步发送消息
     */
    public CompletableFuture<SendResult<String, Message>> sendMessageAsync(String topic, String key, Message message) {
        log.info("异步发送消息到主题: {}, Key: {}", topic, key);
        return kafkaTemplate.send(topic, key, message).completable();
    }
    
    /**
     * 发送批量消息
     */
    public void sendBatchMessages(String topic, int count) {
        log.info("发送批量消息到主题: {}, 数量: {}", topic, count);
        
        for (int i = 1; i <= count; i++) {
            Message message = Message.builder()
                    .messageId("batch-" + i)
                    .topic(topic)
                    .content("批量消息内容 " + i)
                    .messageType("BATCH")
                    .key("batch-key-" + i)
                    .build();
            
            sendMessage(topic, message.getKey(), message);
        }
    }
    
    /**
     * 发送高吞吐量消息（用于性能测试）
     */
    public void sendHighThroughputMessages(String topic, int count) {
        log.info("发送高吞吐量消息到主题: {}, 数量: {}", topic, count);
        
        for (int i = 1; i <= count; i++) {
            Message message = Message.builder()
                    .messageId("throughput-" + i)
                    .topic(topic)
                    .content("高吞吐量消息 " + i)
                    .messageType("THROUGHPUT")
                    .key("throughput-key-" + i)
                    .build();
            
            // 使用异步发送提高吞吐量
            sendMessageAsync(topic, message.getKey(), message);
        }
    }
    
    /**
     * 发送分区消息（指定分区）
     */
    public void sendMessageToPartition(String topic, int partition, String key, Message message) {
        log.info("发送消息到指定分区 - Topic: {}, Partition: {}, Key: {}", topic, partition, key);
        
        ListenableFuture<SendResult<String, Message>> future = kafkaTemplate.send(topic, partition, key, message);
        
        future.addCallback(new ListenableFutureCallback<SendResult<String, Message>>() {
            @Override
            public void onSuccess(SendResult<String, Message> result) {
                log.info("分区消息发送成功 - Topic: {}, Partition: {}, Offset: {}", 
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            }
            
            @Override
            public void onFailure(Throwable ex) {
                log.error("分区消息发送失败 - Topic: {}, Partition: {}, Error: {}", 
                        topic, partition, ex.getMessage(), ex);
            }
        });
    }
} 